#include <carbon/core/optional.h>
#include <carbon/core/thread.h>
#include <carbon/core/latch.h>
#include <carbon/core/barrier.h>
#include <carbon/core/block_queue.h>
#include <carbon/core/queued_thread_pool.h>
#include <carbon/core/error.h>

#include <functional>
#include <queue>
#include <string>
#include <iostream>

#include <stdio.h>
#include <unistd.h>

using carbon::Thread;
using carbon::Latch;
using carbon::Barrier;
using carbon::Optional;
using carbon::RunnableFn;
using carbon::QueuedThreadPool;
using String = std::string;

template<typename T>
using BlockingQueue = carbon::BlockingQueue<T>;

using namespace std;





void test_int_queue_offer(BlockingQueue<long> & que)
{
    long value = 1;
    int count = 0;
    while ( 1 ) {
        bool ok = que.offer(value);
        value += 10;
        if ( ok ) ++count;
        else break;
    }

    cout<<"test_int_queue_offer: "<<count<<endl;
}

void test_int_queue_poll(BlockingQueue<long> & que)
{
    long total = 0;
    int count = 0;
    while (1) {
        auto value = que.poll();
        if ( value.hasValue() ) {
            count += 1;
            total += value.get();
        }
        else break;
    }

    cout<<"test_int_queue_poll: count: "<<count<<", total: "<<total<<endl;
}

void test_int_queue_put(BlockingQueue<long> & que, int max)
{
    long value = 1;
    int count = 0;
    while ( count < max ) {
        bool ok = que.put(value);
        value += 10;
        if ( ok ) ++count;
        else break;
    }

    cout<<"test_int_queue_put: "<<count<<endl;
}

#define THREAD_COUNT 5

void test_producer_and_consuer(BlockingQueue<long> & que1)
{
    Barrier             bar(THREAD_COUNT);
    Latch               lat(THREAD_COUNT);


    auto fn_prod = [&](){ bar.await(); test_int_queue_put(que1, 100); lat.countDown(); };
    auto fn_cons = [&](){ lat.await(); test_int_queue_poll(que1); };

    Thread consumer;
    consumer = Thread(fn_cons, "consumer");

    Thread producer[THREAD_COUNT];
    for (int i = 0; i < THREAD_COUNT; ++i) {
        char tname[32];
        snprintf(tname, 32, "producer-%d", i);
        producer[i] = Thread(fn_prod, tname);
    }

    for (int i = 0; i < THREAD_COUNT; ++i) {
        producer[i].join();
    }

    consumer.join();
}

void test_queued_thread_pool(BlockingQueue<long> & que)
{
    int data_count = 0;
    bool ok;
    do { ok = que.offer(1); if (ok) ++data_count; } while(ok);
    printf("%d data in queue\n", data_count);

    std::atomic_long total {0};

    auto fn  = [&]() { 
        auto value = que.poll();
        if ( !value )  return;

        long old = 0;
        long val = 0;
        do {
            old = total.load();
            val = old + value.get();
        } while ( !total.compare_exchange_strong(old, val));
        printf("executed by thread: %lx\n", pthread_self());
    };

    QueuedThreadPool threads("test", THREAD_COUNT, 1, 5);
    ok = threads.start(); assert(ok);

    int task_count = 0;
    int failed = 0;
    for (int i = 0; i < que.capacity(); ++i ) {
        ok = threads.execute(fn);
        if (ok ) ++task_count;
        else ++failed;
    }
    printf("%d tasks sent, failed: %d, cap: %d\n", task_count, failed, que.capacity());

    sleep(10);

    do { ok = que.offer(1); if (ok) ++data_count; } while(ok);

    for (int i = 0; i < que.capacity(); ++i ) {
        ok = threads.execute(fn);
        if (ok ) ++task_count;
        else ++failed;
    }
    printf("%d data count, %d tasks sent, failed: %d, cap: %d\n", data_count, task_count, failed, que.capacity());

    sleep(10);

    threads.stop();

    printf("total: %ld\n", total.load());
}


int main(int argc, char** argv)
{
    BlockingQueue<long> que1(10000);
    
    test_int_queue_offer(que1);
    test_int_queue_poll(que1);
    test_producer_and_consuer(que1);
    test_queued_thread_pool(que1);

    return 0;
}
